package mr.dividends;

import java.io.IOException;

import mr.dividends.DividendsReducer.Tokens;
import mr.dividends.model.DividendsInputWritable;
import mr.dividends.model.DividendsKeyWritable;
import mr.dividends.model.DividendsValueWritable;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;

public class DividendsMapper extends Mapper<LongWritable, DividendsInputWritable, DividendsKeyWritable, DividendsValueWritable> {
	public final static String START_YEAR = "startYear";
	public final static String END_YEAR = "endYear";
	private DividendsKeyWritable keyWritable = new DividendsKeyWritable();
	private DividendsValueWritable valueWritable = new DividendsValueWritable();
	private int startYear;
	private int endYear;
	
	@Override
	protected void setup(Context context
            ) throws IOException, InterruptedException {
		startYear = context.getConfiguration().getInt(START_YEAR, 1900);
		endYear = context.getConfiguration().getInt(END_YEAR, 2100);
	}
	
	@Override
	protected void map(LongWritable lineNum, DividendsInputWritable inputDividends, Context context) throws IOException, InterruptedException {
		
		int year = Integer.parseInt(inputDividends.getYearStr());
		if(year >= startYear && year <=endYear){
			keyWritable.setYearStr(inputDividends.getYearStr());
			keyWritable.setSymbol(inputDividends.getStockSymbol());
			
			valueWritable.setAnnualDividendsTotal(inputDividends.getDividends());
			valueWritable.setAnnualCount(1);
			context.write(keyWritable, valueWritable);
			
			context.getCounter("Other Stats:", inputDividends.getYearStr()).increment(1);
		}
		else
			context.getCounter("Other Stats:", "Entires Filtered Out").increment(1);
	}
}
